package com.example.cloud.service;

import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.messaging.Message;

import lombok.extern.slf4j.Slf4j;

/**
 * Kafka Consumer<br>
 * Sink用于有单个输入（inbound）通道的应用，通道名称为input
 *
 * @author tongyufu
 */
@Slf4j
@EnableBinding({ Sink.class, SinkSender.class })
public class SinkReceiver {

    /**
     * 监听input通道
     *
     * @param payload 消息
     */
    @StreamListener(Sink.INPUT)
    public void receive(Message<?> payload) {
        log.info("----Received:" + payload);
        //        Acknowledgment acknowledgment = payload.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT,
        //            Acknowledgment.class);
        //        if (acknowledgment != null) {
        //            System.out.println("Acknowledgment provided");
        //            acknowledgment.acknowledge();
        //        }
    }

}
